package com.momo.apachedoris.service.unique;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.momo.apachedoris.dao.vo.DbFieldVO;
import com.momo.apachedoris.dao.vo.DorisSendDataVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.Set;

/**
 * @Author: Jie Li
 * @Date 2022年11月22日 18:30
 * @Description: 使用 Insert 方式同步数据
 * doris版本必须全量字段更新
 * @Version: 1.0
 * <p>Copyright: Copyright (c) 2022</p >
 **/
@Service
@Slf4j
public class DorisJdbcBatchInsertService {
    private static final String JDBC_DRIVER = "com.mysql.jdbc.Driver";
    private static final String DB_URL_PATTERN = "jdbc:mysql://%s:%d/%s?rewriteBatchedStatements=true";
    private static final String HOST = "192.168.189.128"; // Leader Node host
    //    private static final String HOST = "172.19.8.34"; // Leader Node host
    private static final int PORT = 9030;   // query_port of Leader Node
    private static final String DB = "example_db";
    private static final String TBL = "unique_detail";
    private static final String USER = "root";
    private static final String PASSWD = "123qwe";

    private static final int INSERT_BATCH_SIZE = 10000;

    public static void main(String[] args) {


        List<List<DbFieldVO>> listList = Lists.newArrayList();
        List<DbFieldVO> dbFieldVOS = Lists.newArrayList();
        dbFieldVOS.add(new DbFieldVO("city", "苏州3"));
        dbFieldVOS.add(new DbFieldVO("age", 28));
        dbFieldVOS.add(new DbFieldVO("sex", 1));
        dbFieldVOS.add(new DbFieldVO("phone", 1));
        dbFieldVOS.add(new DbFieldVO("address", "地址11"));
        dbFieldVOS.add(new DbFieldVO("user_id", 2022));
        dbFieldVOS.add(new DbFieldVO("username", "张三3"));
        dbFieldVOS.add(new DbFieldVO("register_time", "2022-11-22 18:54:55"));
        listList.add(dbFieldVOS);

        DorisSendDataVO sendDataVO = new DorisSendDataVO();
        sendDataVO.setLabel("label6");
        batchInsert(listList, sendDataVO);


        JSONArray streamLoadVO = new JSONArray();
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("city", "苏州34");
        jsonObject.put("age", 28);
        jsonObject.put("sex", 1);
        jsonObject.put("phone", 1);
        jsonObject.put("address", "地址11");
        jsonObject.put("user_id", 2022);
        jsonObject.put("username", "张三3");
        jsonObject.put("register_time", "2022-11-22 18:54:55");
        streamLoadVO.add(jsonObject);
        sendDataVO.setLabel("label5");

//        batchInsert(streamLoadVO, sendDataVO);
    }


    /**
     * doris版本必须全量字段更新
     * 批量插入数据
     * 序列化数据的时候  JSONObject.toJSONString(业务数据, SerializerFeature.WriteNullStringAsEmpty)
     *
     * @param streamLoadVO
     * @param sendDataVO
     */
    private static void batchInsert(JSONArray streamLoadVO, DorisSendDataVO sendDataVO) {

        StringBuilder fields = new StringBuilder();
        StringBuilder values = new StringBuilder();
        fields.append(" (");
        values.append(" values(");


        JSONObject jsonObject = streamLoadVO.getJSONObject(0);
        Set<String> fieldKey = jsonObject.keySet();
        int ii = 0;
        for (String field : fieldKey) {
            fields.append(field);
            values.append("?");
            if (ii < jsonObject.size() - 1) {
                fields.append(",");
                values.append(",");
            }
            ii++;
        }

        fields.append(") ");
        values.append(") ");
        // 注意末尾不要加 分号 ";"
        String label = " ";
        if (StringUtils.isNotBlank(sendDataVO.getLabel())) {
            label = " WITH LABEL " + sendDataVO.getLabel();
        }
        String query = "insert into " + TBL + label + fields + values;

        // 设置 Label 以做到幂等。
        // String query = "insert into " + TBL + " WITH LABEL my_label values(?, ?)";

        Connection conn = null;
        PreparedStatement stmt = null;
        String dbUrl = String.format(DB_URL_PATTERN, HOST, PORT, DB);
        try {
            Class.forName(JDBC_DRIVER);
            conn = DriverManager.getConnection(dbUrl, USER, PASSWD);
            stmt = conn.prepareStatement(query);

            for (int i = 0; i < streamLoadVO.size(); i++) {
                JSONObject objItem = streamLoadVO.getJSONObject(i);
                Set<String> objKey = objItem.keySet();
                int fieldIndex = 0;
                for (String key : objKey) {
                    stmt.setObject(fieldIndex + 1, objItem.get(key));
                    fieldIndex++;
                }
                stmt.addBatch();

            }


            int[] res = stmt.executeBatch();
            System.out.println(res.length);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (stmt != null) {
                    stmt.close();
                }
            } catch (SQLException se2) {
                se2.printStackTrace();
            }
            try {
                if (conn != null) conn.close();
            } catch (SQLException se) {
                se.printStackTrace();
            }
        }
    }

    /**
     * doris版本必须全量字段更新
     * 手动批量插入数据
     *
     * @param listList
     * @param sendDataVO
     */
    private static void batchInsert(List<List<DbFieldVO>> listList, DorisSendDataVO sendDataVO) {


        StringBuilder fields = new StringBuilder();
        StringBuilder values = new StringBuilder();
        fields.append(" (");
        values.append(" values(");
        for (int i = 0; i < listList.get(0).size(); i++) {
            DbFieldVO dbFieldVO = listList.get(0).get(i);
            fields.append(dbFieldVO.getField());
            values.append("?");
            if (i < listList.get(0).size() - 1) {
                fields.append(",");
                values.append(",");
            }
        }
        fields.append(") ");
        values.append(") ");
        // 注意末尾不要加 分号 ";"
        String label = " ";
        if (StringUtils.isNotBlank(sendDataVO.getLabel())) {
            label = " WITH LABEL " + sendDataVO.getLabel();
        }
        String query = "insert into " + TBL + label + fields + values;

        // 设置 Label 以做到幂等。
        // String query = "insert into " + TBL + " WITH LABEL my_label values(?, ?)";

        Connection conn = null;
        PreparedStatement stmt = null;
        String dbUrl = String.format(DB_URL_PATTERN, HOST, PORT, DB);
        try {
            Class.forName(JDBC_DRIVER);
            conn = DriverManager.getConnection(dbUrl, USER, PASSWD);
            stmt = conn.prepareStatement(query);

            for (List<DbFieldVO> dbFieldVOS : listList) {
                for (int i = 0; i < dbFieldVOS.size(); i++) {
                    DbFieldVO dbFieldVO = dbFieldVOS.get(i);
                    stmt.setObject(i + 1, dbFieldVO.getValue());
                }
                stmt.addBatch();
            }

            int[] res = stmt.executeBatch();
            System.out.println(res.length);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (stmt != null) {
                    stmt.close();
                }
            } catch (SQLException se2) {
                se2.printStackTrace();
            }
            try {
                if (conn != null) conn.close();
            } catch (SQLException se) {
                se.printStackTrace();
            }
        }
    }


}
